package jobs.job4.wcdemo;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/*
1.统计输入文本中每个单词出现的次数
2.如果输入文本为1行，则直接进行统计。
3.如果输入文本为n行，则将输入分成2个子任务，第一个子任务的输入为第一行，第二个子任务的输入为剩下的n-1行
 */
public class WordCountTask extends RecursiveTask<Map<String, Integer>> {
    String txt;//输入文本
    public WordCountTask(String txt) {
        this.txt = txt.trim();
    }
    @Override
    protected Map<String, Integer> compute() {

        int index = txt.indexOf('\n');

        if (index > 0) {
//            说明输入为多行
//  输入拆分算法：将输入拆分成两个2任务，基本任务和大任务，就是 第一行 和 剩下的行
            WordCountTask baseTask = new WordCountTask(txt.substring(0, index));
            WordCountTask LeftTask = new WordCountTask(txt.substring(index + 1));

//          将两个任务分别加入线程池
            LeftTask.fork();
            baseTask.fork();

//          阻塞直到两个任务都执行完成
            Map<String, Integer> baseResult = baseTask.join();
            Map<String, Integer> leftResult = LeftTask.join();


//            聚合两个任务的结果
            Map<String, Integer> result = new HashMap<>(leftResult);
            for (Map.Entry<String, Integer> entry : baseResult.entrySet()) {
                result.compute(entry.getKey(), (key, count) -> {
                    if (count == null) count = 0;
                    return count + entry.getValue();//返回单词在两个任务中的统计和。
                });
            }

            return result;      // 返回聚合结果
        } else {
//            说明输入为单行文本，为基本任务，不可分割，可直接统计
            String[] words = txt.split("\\s++");//按空白字符分割单词
            Map<String, Integer> map = new HashMap<>();
            for (String word : words) // 遍历所有单词，统计出现次数
                map.compute(word, (key, count) -> {
                    if (count == null) count = 0;
                    return count + 1;//单词每出现一次，统计加1
                });
            return map; // 返回单行文本的统计结果
        }
    }

    public static void main(String[] args) throws FileNotFoundException {
        StringBuilder txt = new StringBuilder();
//        读取文件中全部内容
        try (Scanner scanner = new Scanner(new File("jobinputs/job4/words.txt"))) {
            while (scanner.hasNext()) {
                String line = scanner.nextLine();
                txt.append(line).append('\n');
            }
        }
        //创建单词统计任务
        WordCountTask wordCountTask = new WordCountTask(txt.toString());
        //创建Fork/Join 线程池，使用默认的并行度
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //将大任务加入线程池执行，并阻塞直到得到统计结果
        Map<String, Integer> result = forkJoinPool.invoke(wordCountTask);
//     打印输出结果
        System.out.println(result);
//        {hive=1, bigdata=1, mapreduce=3, test=1, lagou=3, hdfs=2, hadoop=2, yarn=3}

    }

}
